package com.pie4cloud.pie.bus.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.pie4cloud.pie.bus.api.dto.ConsumerDto;
import com.pie4cloud.pie.bus.api.dto.MessageDto;
import com.pie4cloud.pie.bus.api.entity.DataKafConsumer;
import com.pie4cloud.pie.bus.config.FeignRemote;
import com.pie4cloud.pie.bus.service.KafkaManageService;
import com.pie4cloud.pie.bus.service.KafkaSendService;
import com.pie4cloud.pie.common.core.util.R;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.List;

@Service
@RequiredArgsConstructor
@Slf4j
public class KafkaSendServiceImpl implements KafkaSendService {
	@Autowired
	private KafkaTemplate kafkaTemplate;
	@Autowired
	private FeignRemote feignRemote;
	private final KafkaManageService kafkaManageService;


	public R sendKafka(MessageDto messageDto){
		try {
			kafkaTemplate.send(messageDto.getTopic(), messageDto.getContent());
		} catch (Exception e) {
			return R.failed();
		}
		return R.ok();
	}

	public R sendKafkas(List<MessageDto> messageDtos){
		try {
			for(MessageDto messageDto:messageDtos){
				kafkaTemplate.send(messageDto.getTopic(), messageDto.getContent());
			}
		} catch (Exception e) {
			return R.failed();
		}
		return R.ok();
	}

	@Override
	public R sendConsumer(ConsumerDto consumerDto) {
		try {
			kafkaManageService.deleteByTopicAndGroup(consumerDto.getTopic(),consumerDto.getKafGroup());
			DataKafConsumer dataKafConsumer=new DataKafConsumer();
			BeanUtil.copyProperties(consumerDto,dataKafConsumer);
			Date date=new Date();
			String time=String.valueOf(date.getTime());
			consumerDto.setTime(time);
			kafkaManageService.send(consumerDto);
			dataKafConsumer.setConsumptionTime(date);
			kafkaManageService.save(dataKafConsumer);
		} catch (Exception e) {
		    return R.failed();
		}
		return R.ok();
	}



}
